package org.egomsl.mw.table;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.egomsl.mw.record.Record;
import org.egomsl.mw.record.RecordScanner;
import org.egomsl.mw.region.RegionReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;

public class TableReader{
    private static Logger log = LoggerFactory.getLogger(TableReader.class);

    private FileSystem fs;
    private TableDesc tableDesc;

    private ArrayList<RegionReader> regionReaders;

    public TableReader(FileSystem fs, Path tableDir) throws IOException {
        this.fs = fs;
        this.tableDesc = TableDesc.loadFromFs(fs, tableDir);

        regionReaders = new ArrayList<>();
        for(Path regionDir : tableDesc.getPartitionDirs()) {
            regionReaders.add(new RegionReader(fs, regionDir));
        }
    }

    public TableDesc getTableDesc() {
        return tableDesc;
    }

    public ArrayList<RegionReader> getRegionReaders() {
        return regionReaders;
    }

    public RecordScanner getScanner() {
        return new TableRecordScanner(regionReaders);
    }

    private static class TableRecordScanner implements RecordScanner {
        private RecordScanner regionRecordScanner;
        private Iterator<RegionReader> regionReaders;

        public TableRecordScanner(ArrayList<RegionReader> regionReaders) {
            this.regionReaders = regionReaders.iterator();
            this.regionRecordScanner = null;
        }

        @Override
        public boolean hashNext() {
            if(regionRecordScanner != null && regionRecordScanner.hashNext()) {
                return true;
            }else {
                if(regionReaders.hasNext()){
                    regionRecordScanner = regionReaders.next().getScanner(true);
                    if(regionRecordScanner.hashNext()) return true;
                }
            }
            return false;
        }

        @Override
        public Record next() {
            return regionRecordScanner.next();
        }
    }
}
